#include <errno.h>
#include <fcntl.h>
#include <glib.h>
#include <libxml/parser.h>
#include <libxml/tree.h>
#include <libxml/xpath.h>
#include <pthread.h>
#include <semaphore.h>
#include <stdlib.h>
#include <stdio.h>
#include <syslog.h>

#include "config.h"
#include "sms_debug.h"
#include "sms_basic_types.h"
#include "sms_router.h"

#define PROGRAM_NAME "smsrouter"
#define SEND_QUEUE_STABLE_TIME 5
#define SEND_SEMA_SPEC 5

typedef enum 
{
    STATE_TODO,
    STATE_ONGO,
    STATE_DONE,
    STATE_FAIL,
} TASK_STATES;


DEFINE_QUEUE_FNS(Sender, SMS_TASK, QUEUE_MAX_LEN)    


void out_queue_insert(SENDER_PARAMETER *sender,                     
                      SMS_TASK *task)                             
{                                                                       
    unsigned int inserted = FALSE;                                      
    do                                                                 
    {    
        pthread_mutex_lock(&sender->out_queue_lock);
        // reset timer every time inserted a element to the queue.
        g_timer_reset(sender->timer);                
        inserted = MKSSenderQueue_Insert(&sender->out_queue, task);
        pthread_mutex_unlock(&sender->out_queue_lock);              
    } while (!inserted);                                                
    sem_post(sender->out_queue_sema);                             
}


gboolean sender_add_filter(SENDER_PARAMETER *sender, char *string)
{
    ENTRY;
    GError *error = NULL;
    if (sender == NULL || string == NULL)
    {
        LDEBUG(0, ("sender/string is empty, cannot add filter"));
        RETURN(FALSE);
    }
    sender->filter = g_regex_new(string, 0, 0, &error);
    if (sender->filter == NULL)
    {
        g_error(error->message);
        RETURN(FALSE);
    }
    RETURN(TRUE);
}

gboolean send_message(GPtrArray *senders, SMS_TASK *task)
{
    int index = 0;
    gboolean result;
    GError *error = NULL;
    GMatchInfo *match_info;
    for (index; index < senders->len; index ++ )
    {
       SENDER_PARAMETER *sender = g_ptr_array_index(senders, index);
	    if (!sender->filter)
	    {	
	        g_error(error->message);
	        continue;
	    }
      //  result = g_regex_match_all_full(sender->filter, task->cell_phone_number, strlen(task->cell_phone_number), 0, G_REGEX_MATCH_ANCHORED, NULL, &error);
      //  if (result) // matched
      //  {
            out_queue_insert(sender, task);
      //      break;
      //  }
    }
}


GArray *parse_xml(xmlNodePtr root)
{
    GArray *array = g_array_new(FALSE, TRUE, sizeof(SMS_TASK));
    xmlChar *szKey = NULL;
    xmlNodePtr node = NULL;
    root = root->xmlChildrenNode;
    while (root)
    {
        SMS_TASK task = {0};
        xmlAttrPtr attrPtr = root->properties;
        if (attrPtr)
        {
            if (!xmlStrcmp(attrPtr->name, BAD_CAST "type"))
            {
               xmlChar* szAttr = xmlGetProp(root, BAD_CAST "type");
               task.type = atoi(szAttr);
               xmlFree(szAttr);
           }
        }
        node = root->xmlChildrenNode;
        if (!node) // skip some blank node
        {
            root = root->next;
            continue;
        }
        while (node)
        {
            if ((!xmlStrcmp(node->name, (const xmlChar *)"number")))
            {
                szKey = xmlNodeGetContent(node);
                task.cell_phone_number = g_malloc0(strlen(szKey) + 1);
                g_stpcpy(task.cell_phone_number, szKey);
                xmlFree(szKey);
            } else if ((!xmlStrcmp(node->name, (const xmlChar *)"message")))
            {
                szKey = xmlNodeGetContent(node);
                task.message = g_malloc0(strlen(szKey) + 1);
                strcpy(task.message, szKey);
                xmlFree(szKey);
            }
            node = node->next;
        }
        fprintf(stderr, "parse number is %s, message is %s\n", task.cell_phone_number, task.message);
        g_array_append_val(array, task);
        root = root->next;
    }
    return array;
}


SENDER_PARAMETER *sender_para_init()
{
    ENTRY;
    GRand *rand = g_rand_new_with_seed(1234567);
    // init mutex, semephone,
    SENDER_PARAMETER *sender = g_new0(SENDER_PARAMETER, 1);
    MKSSenderQueue_Init(&sender->out_queue);
    sender->sema_spec.tv_sec = SEND_SEMA_SPEC;
 
    if (!(sender->timer = g_timer_new()))
    {
        LERROR(0, ("create sender timer erorr"));
        goto error;
    }
    gchar *sema_name = g_strdup_printf("%i", g_rand_int_range(rand, 0.0, 100000));
    while ((sender->out_queue_sema = sem_open(sema_name, O_CREAT, 777, 0)) == SEM_FAILED) 
    {
        sema_name = g_strdup_printf("%i", g_rand_int_range(rand, 0.0, 100000));
    }
   
    LDEBUG(0, ("init semephone name %s", sema_name)); 
    if (pthread_mutex_init(&sender->out_queue_lock, NULL)) 
    {
        fprintf(stderr, "create sender out queue mutex error");
        goto error;
    }
    goto done;
error:
    if (sender->timer) 
    {
        g_timer_destroy(sender->timer);
    }
    if (sender->out_queue_sema)
    {
        sem_close(sender->out_queue_sema);
    }
    g_free(sender);
    return NULL;
done:
    RETURN(sender);
}


void sms_task_free_content(SMS_TASK *task)
{
    // this function free cell_phone_number and message but not free task itself.
    // because task maybe not dynamic allocated.
    g_free(task->cell_phone_number);
    g_free(task->message);
}

void sender_cleanup_callback(void *parameters)
{
    SENDER_PARAMETER *sender = parameters;
    sem_close(sender->out_queue_sema);
    pthread_mutex_destroy(&sender->out_queue_lock);
    g_regex_unref(sender->filter);
    g_timer_destroy(sender->timer);
    g_free(sender);    
}


void *send_thread(void *parameters)
{
    unsigned int removed = FALSE;
    int send_result = 0;
    SENDER_PARAMETER *sender = parameters;
    GString *cell_phone_number = g_string_new(NULL);
    GString *message = g_string_new(NULL);

    SMS_TASK task = {0};
    gdouble time_elapsed = 0.0f;
    pthread_cleanup_push(&sender_cleanup_callback, sender);

    g_timer_start(sender->timer);

    while (1) 
    {
        time_elapsed =  g_timer_elapsed(sender->timer, NULL);
        if (!(time_elapsed > SEND_QUEUE_STABLE_TIME || MKSSenderQueue_Full(&sender->out_queue)))
        {
            // continue to wait for queue stable.
            continue;
        }
        while(!MKSSenderQueue_Empty(&sender->out_queue))
        {
            pthread_mutex_lock(&sender->out_queue_lock);
            removed = MKSSenderQueue_Remove(&sender->out_queue, &task);
            if (removed) {
               switch (task.type) {
                  case PACKET_TYPE_SMS_NORMAL:
                      // combine all single message to one package.
                      fprintf(stdout, "number is %s, message is %s\n", task.cell_phone_number, task.message);
                      g_string_append(cell_phone_number, task.cell_phone_number);
                      g_string_append(message, task.message);
                      if (!MKSSenderQueue_Empty(&sender->out_queue))
                      {
                           g_string_append_c(cell_phone_number, ',');
                           g_string_append_c(message, ',');
                      }
                  break;
                  case PACKET_TYPE_SMS_GROUP:
                      assert(sender->send_grouped_messages != NULL);
                      send_result = sender->send_grouped_messages(sender->sn, sender->password, task.cell_phone_number, task.message);
                      fprintf(stdout, "number is %s, message is %s\n", task.cell_phone_number, task.message);
                      syslog(LOG_INFO, "event = groupedsend; number = %s; message = %s; result = %d\n", 
                              task.cell_phone_number, task.message, send_result);
                  default:
                  break; 
               }
            }
            sms_task_free_content(&task);
            pthread_mutex_unlock(&sender->out_queue_lock);
        }
        if (cell_phone_number->len && message->len) {
            assert(sender->send_grouped_individual_messages != NULL);
            send_result = sender->send_grouped_individual_messages(sender->sn, 
                                                                   sender->password, 
                                                                   cell_phone_number->str, 
                                                                   message->str);
            syslog(LOG_INFO, "event = send; number = %s; message = %s\n result = %d", 
                   cell_phone_number->str, message->str, send_result);
            
            fprintf(stderr, "event = send; number = %s; message = %s\n result = %d", 
                    cell_phone_number->str, message->str, send_result);
            //lsyslog("event = send, number = %s, message = %s, result = %d\n", cell_phone_number->str, message->str, send_result);
        }
        g_string_erase(cell_phone_number, 0, cell_phone_number->len);
        g_string_erase(message, 0, message->len);
        sem_timedwait(sender->out_queue_sema, &sender->sema_spec); 
    }
    g_string_free(cell_phone_number, TRUE);
    g_string_free(cell_phone_number, FALSE);
    pthread_cleanup_pop(0);
}


void *recv_thread(void *receiver_parameter)
{
    RECEIVER_PARAMETER *param = (RECEIVER_PARAMETER *)receiver_parameter;
    TASK_STATES state = STATE_TODO;
    char recv_buf[NET_BUF_LEN];
    char send_buf[NET_BUF_LEN];
    int server_sockfd = param->cli_sockfd;
    int received_length = 0;
    int remained_length = 0;
    int first_packet_length = 0;

    SMS_PACKET_HEADER *header = NULL;

    xmlChar *body = NULL;
    xmlChar *task = NULL;
    xmlChar *write_point = NULL;
    xmlDocPtr doc = NULL;
    xmlNodePtr root = NULL;

    while (state != STATE_FAIL)
    {
        fprintf(stderr, "state is %d\n", state);
        switch (state)
        {
            case STATE_TODO: // get packet header
                // if receive error or remote close the socket, then exit the thread.
                if ((received_length = recv(server_sockfd, recv_buf, NET_BUF_LEN, 0)) <= 0)
                {
                    state = STATE_FAIL;
                    continue;
                }
                fprintf(stderr, "received data is %s, length is %d", 
                        recv_buf, received_length);
                header = recv_buf;
                body = (xmlChar *)malloc(header->packet_size);
                write_point = body;
                first_packet_length = received_length - sizeof(SMS_PACKET_HEADER);
                memcpy(body, recv_buf + sizeof(SMS_PACKET_HEADER), first_packet_length);
                write_point += first_packet_length;
                remained_length = header->packet_size - first_packet_length;
                if (remained_length > 0)
                {
                    state = STATE_ONGO;
                    continue;
                } 
                else
                {
                    state = STATE_DONE;
                    continue;
                }
                break;
            case STATE_ONGO: // get xml stream
                fprintf(stderr, "try to receive data");
                if ((received_length = recv(server_sockfd, recv_buf, NET_BUF_LEN, 0)) < 0 )
                {
                    state = STATE_FAIL;
                }

                fprintf(stderr, "receive data");
                memcpy(write_point, recv_buf, received_length);
                if (received_length < remained_length) // more than one recv 
                {
                    write_point += received_length;        
                }
                else 
                {
                    state = STATE_DONE;
                }
                memset(recv_buf, 0, NET_BUF_LEN);
                break;
            case STATE_DONE:
                fprintf(stderr, "received data is %s", body);
                doc = xmlRecoverDoc(body);
                if (!doc)
                {
                    printf("can not parse stream received.");
                }

                GArray *array = parse_xml(xmlDocGetRootElement(doc));
                int i = 0;
                for (i; i < array->len; i++)
                {
                    SENDER_PARAMETER *sender = NULL;
                    SMS_TASK *task = &g_array_index(array, SMS_TASK, i);   
                    // give data to proper sender
		              send_message(param->senders, task);
                }
                g_array_free(array, TRUE);
                xmlFreeDoc(doc);
                free(body);
                state = STATE_TODO;
                break;
            default:
                break;           
        }
    }
    // error happened, close the socket and quick the thread.
    pthread_exit(NULL);
}
